package com.zx.learn.flink.apps.app1mysqltomysql;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.*;

/**
 * MySql源
 * implements ResultTypeQueryable<T>
 * @author Ken.Zheng
 * @date 2022/12/02
 **/
@Slf4j
public abstract class MySQLSource<T> extends RichSourceFunction<T> {

    PreparedStatement ps;

    Connection connection;

    abstract Connection getConnection();

    abstract String getSql();

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        connection = getConnection();
        if (connection != null) {
            ps = this.connection.prepareStatement(getSql());
        }
        log.debug("获取jdbc连接:{},{}", connection, ps);
    }

    @Override
    public void cancel() {

    }

    @Override
    public void close() throws Exception {
        super.close();
        //关闭连接和释放资源
        if (connection != null) {
            connection.close();
        }
        if (ps != null) {
            ps.close();
        }
    }

//    @Override
//    public TypeInformation<T> getProducedType() {
//        return SampleUser.class;
//    }
}
